package com.tony.reactiveStream;

import java.util.List;

/**
 * @version V1.0  </br>
 * @ClassName MyReactiveAppWithProcessor </br>
 * @Description TODO  </br>
 * @Author tonywang</br>
 * @DATE 3/14/2019 5:31 PM</br>
 */
public class MyReactiveAppWithProcessor {

    public static void main(String[] args) throws InterruptedException {
        // Create End Publisher
        MyPublisher publisher = new MyPublisher();
        // Create Processor
        MyProcessor transformProcessor = new MyProcessor(s-> {return new Freelancer(s.getId(), s.getId() + 100, s.getName());});
        //Create End Subscriber
        MyFreelancerSubscriber subs = new MyFreelancerSubscriber();
        //Create chain of publisher, processor and subscriber
        publisher.subscribe(transformProcessor);
        transformProcessor.subscribe(subs);
        List<Employee> emps = EmpHelper.getEmps();
        // Publish items
        System.out.println("Publishing Items to Subscriber");
        emps.stream().forEach(i -> publisher.submit(i));
        // Logic to wait for messages processing to finish
        while (emps.size() != subs.getCounter()) {
            Thread.sleep(10);
        }
        // Closing publishers
        publisher.close();
        transformProcessor.close();

        System.out.println("Exiting the app");
    }

}
